package org.project.order

import org.FlinkStreamApp
import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector
import org.project.bean.OrderEvent

/**
 * description ：订单超时, 不使用 cep 实现
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/27 13:56
 * modified By ：
 * version:    : 1.0
 */
object OrderTimeoutWithOutCep extends FlinkStreamApp {
  override def doSomeThing(environment: StreamExecutionEnvironment) = {
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // 创建订单数据
    val source = environment.fromElements(
      OrderEvent("1", "create", "1558430842"),
      OrderEvent("2", "create", "1558430843"),
      OrderEvent("2", "pay", "1558430844"),
      OrderEvent("3", "pay", "1558430942"),
      OrderEvent("4", "pay", "1558430943")
    ).assignAscendingTimestamps(_.eventTime.toLong * 1000)
      .keyBy(_.orderId)

    // cep 不使用实现超时提醒
    source
      .process(new MyProcess())
      .print()
  }

  class MyProcess extends KeyedProcessFunction[String, OrderEvent, String] {
    // 保存订单状态
    lazy val orderState = getRuntimeContext.getState(
      new ValueStateDescriptor[OrderEvent]("orderList", Types.of[OrderEvent])
    )

    override def processElement(value: OrderEvent, ctx: KeyedProcessFunction[String, OrderEvent, String]#Context, out: Collector[String]) = {
      // 如果订单状态为创建，并且 orderState 为空
      if (value.eventType == "create" && null == orderState.value) {
        // 将 orderState 状态设置了当前订单
        orderState.update(value)
        // 设置一个定时器，为 5s
        ctx.timerService().registerEventTimeTimer(value.eventTime.toLong * 1000 + 5 * 1000)
      }

      // 如果订单状态为 支付，就清空 orderState
      if (value.eventType == "pay" && null != orderState.value()) {
        orderState.clear()
      }
    }

    // 定时器
    override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, OrderEvent, String]#OnTimerContext, out: Collector[String]) = {
      if (null != orderState.value) {
        out.collect(s"订单ID为： ${orderState.value().orderId} 5s以内没有支付")
        orderState.clear()
      }
    }
  }

}
